// -*-mode:c++; coding:utf-8-*-

#include "logger.hpp"
#include "naming_strategy.hpp"

#include "scanner.hpp"

namespace xbase
{
	void scanner::on_get_meta(location_cache_value meta, const error_code& ec)
	{
		if (!ec)
		{
			m_meta_data = meta;

			LOG(TRACE, "tablet: " << m_meta_data.tablet_info_p->m_table_name
				<< " [" << dump_string_hex(m_meta_data.tablet_info_p->m_start_row) << ","
				<< dump_string_hex(m_meta_data.tablet_info_p->m_end_row) << "]"
				<< " is at " << m_meta_data.server_addr_p->m_hostname << ":"
				<< m_meta_data.server_addr_p->m_port
				);

			endpoint_type ep(
				boost::asio::ip::address::from_string(m_meta_data.server_addr_p->m_hostname),
				m_meta_data.server_addr_p->m_port
				);

			std::string scan_start_row = m_cur_row;
			// 将scan_end_row设为较小的行名
			std::string scan_end_row = m_meta_data.tablet_info_p->m_end_row;
			if (!m_end_row.empty()
				&& (scan_end_row.empty() || m_end_row<scan_end_row))
			{
				scan_end_row = m_end_row;
			}

			std::string tablet_name = naming_strategy::tablet_name(
				m_meta_data.tablet_info_p->m_tablet_id, m_table_name,
				m_meta_data.tablet_info_p->m_start_row, m_meta_data.tablet_info_p->m_end_row
				);

			m_tablet_scan_info.m_tablet_name = tablet_name;
			m_tablet_scan_info.m_start_row = scan_start_row;
			m_tablet_scan_info.m_end_row = scan_end_row;

			do_new_tablet_scan(ep);
		}
		else
		{
			if (ec == xbase::error::tablet_not_exists)
			{
				on_fail(ec);
			}
			else
			{
				m_stopped = true;
				dispatch(row_result_ptr(), ec);
			}
		}
	}

	void scanner::on_get_next_row(row_result::ptr_type row_ptr, const error_code &ec)
	{
		if (!ec)
		{
			m_retry_times = 0;

			// 如果扫描中出错，重新扫描需要以原来扫描过的行作为首行，所以返回的第一行会重复
			if (m_include_cur_row
				|| row_ptr->m_row_name != m_cur_row)
			{
				m_cur_row = row_ptr->m_row_name;
				dispatch(row_ptr, ec);
			}
			else
			{
				m_include_cur_row = true;
				do_async_get_next_row();
			}
		}
		else
		{
			if (ec == xbase::error::eos)
			{
				m_retry_times = 0;

				if (m_tablet_scan_info.m_end_row.empty()
					|| m_end_row == m_tablet_scan_info.m_end_row)
				{
					// 已经扫描到了表尾，扫描结束
					m_stopped = true;
					dispatch(row_result_ptr(), ec);
				}
				else
				{
					// 找下一个子表的位置
					// 下一个子表的起始行就是当前扫描完的子表的结束行
					do_location_cache_search(m_tablet_scan_info.m_end_row, false);
				}
			}
			else
			{
				// 重试
				on_fail(ec);
			}
		}
	}

	void scanner::do_retry()
	{
		// 扫描当前子表发生了错误，需要重新查找子表信息
		// m_cur_row是之前已经返回的行

		// 如果本次扫描已经返回了一定的行，则确保m_cur_row已经返回
		// 否则，m_cur_row是否已经返回取决于原来的标志
		if (m_tablet_scanner_ptr->get_returned_row_count() > 0)
		{
			m_include_cur_row = false;
		}

		do_location_cache_search(m_cur_row, true);
	}

	void scanner::on_fail(const error_code &ec)
	{
		if(m_options.can_retry(m_retry_times))
		{
			LOG(WARN, "tablet " << m_table_name
				<< "[" << dump_string_hex(m_tablet_scan_info.m_start_row)
				<< " , " << dump_string_hex(m_tablet_scan_info.m_end_row)
				<< "] m_cur_row: " << dump_string_hex(m_cur_row)
				<< " retry: ec = " << ec << " - " << ec.message()
				<< ", retry_times = " << m_retry_times);

			++ m_retry_times;
			if(!m_timer)
				m_timer.reset(new boost::asio::deadline_timer(this->get_lower_level_service()));

			m_timer->expires_from_now(m_options.get_retry_interval());
			m_timer->async_wait(boost::bind(&this_type::do_retry, this_ptr()));
		}
		else
		{
			on_error(ec);
		}
	}

	void scanner::on_error(const error_code &ec)
	{
		LOG(ERROR, "tablet[" << dump_string_hex(m_tablet_scan_info.m_start_row) << " , "
			<< dump_string_hex(m_tablet_scan_info.m_end_row)
			<< "] error: " << ec << " - " << ec.message()
			<< ", retry_times = " << m_retry_times);

		m_retry_times = 0;

		m_stopped = true;
		dispatch(row_result_ptr(), ec);
	}

	void scanner::do_new_tablet_scan(const endpoint_type &ep)
	{
		LOG(DEBUG, "create scanner, tablet_name: " << m_tablet_scan_info.m_tablet_name
			<< " scan_start_row: " << dump_string_hex(m_tablet_scan_info.m_start_row)
			<< " scan_end_row: " << dump_string_hex(m_tablet_scan_info.m_end_row)
			);

		m_tablet_scanner_ptr = tablet_scanner_ptr(
			new tablet_scanner(m_io_service_pool, ep, m_session_manager, m_tablet_scan_info)
			);

		do_async_get_next_row();
	}

	void scanner::do_async_get_next_row()
	{
		LOG(TRACE, "do_async_get_next_row");
		m_tablet_scanner_ptr->async_get_next_row(
			boost::bind(&this_type::on_get_next_row, this_ptr(), _1, _2)
			);
	}

	void scanner::do_location_cache_search(std::string &request_row, bool hard)
	{
		LOG(INFO, "search in location cache, request_row: " << dump_string_hex(request_row)
			<< " hard: " << hard);

		m_location_cache.async_get_meta(
			boost::bind(&this_type::on_get_meta, this, _1, _2),
			m_table_name, request_row, hard, m_cur_meta_got_time
			);
	}

	void scanner::dispatch(row_result_ptr row_ptr, error_code ec)
	{
		LOG(TRACE, "dispatch row: " << ((!row_ptr) 
			? std::string("#<null row result ptr>")
			: dump_string_hex(row_ptr->m_row_name)));

		this->get_upper_level_service().dispatch(boost::bind(m_handler, row_ptr, ec));
	}
}
